package org.flown.util.mq;


import jakarta.jms.*;

import java.io.Serializable;
import java.util.HashMap;


/**
 * <bean id="MQUtil" class="org.flown.util.mq.MQUtil">
 *     <property name="TopicConnectionFactory" ref="TopicConnectionFactory"/>
 * <bean/>
 *
 * @author weifei(flown)
 * */
public class MQUtil
{
    private static boolean autoCloseHook = false;

    protected static TopicConnectionFactory CONNECTION_FACTORY;
    protected static TopicConnection CONNECTION;
    protected static TopicSession SESSION;
    protected static final HashMap<String, Destination> DESTINATIONS = new HashMap<>();
    protected static final HashMap<String, MessageProducer> MESSAGE_PRODUCERS = new HashMap<>();
    protected static final HashMap<String, MessageConsumer> MESSAGE_CONSUMERS = new HashMap<>();
    protected static final HashMap<String, Topic> TOPICS = new HashMap<>();
    protected static final HashMap<String, TopicPublisher> TOPIC_PUBLISHERS = new HashMap<>();
    protected static final HashMap<String, TopicSubscriber> TOPIC_SUBSCRIBERS = new HashMap<>();


    protected MQUtil(){}
    public MQUtil(TopicConnectionFactory topic_connection_factory)
    {
        try {this.setTopicConnectionFactory(topic_connection_factory);}
        catch (JMSException e) {throw new RuntimeException(e);}
    }

    public void setTopicConnectionFactory(TopicConnectionFactory factory)
            throws JMSException
    {
        MQUtil.CONNECTION_FACTORY = factory;
        MQUtil.CONNECTION = MQUtil.CONNECTION_FACTORY.createTopicConnection();
        MQUtil.CONNECTION.start();
        MQUtil.SESSION = MQUtil.CONNECTION.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);

        if (!MQUtil.autoCloseHook)
        {
            Runtime.getRuntime().addShutdownHook(new Thread(MQUtil::close));
            MQUtil.autoCloseHook = true;
        }
    }


    public static void init(TopicConnectionFactory topic_connection_factory)
    {
        new MQUtil(topic_connection_factory);
    }


    public static void close()
    {
        try
        {
            for (MessageProducer p: MQUtil.MESSAGE_PRODUCERS.values())
                p.close();
            for (MessageConsumer c: MQUtil.MESSAGE_CONSUMERS.values())
                c.close();

            for (TopicPublisher tp: MQUtil.TOPIC_PUBLISHERS.values())
                tp.close();
            for (TopicSubscriber ts: MQUtil.TOPIC_SUBSCRIBERS.values())
                ts.close();

            MQUtil.SESSION.close();
            MQUtil.CONNECTION.close();
        }
        catch (JMSException e) {throw new RuntimeException(e);}
    }


    /*create message*/
    /*创建消息*/
    public static BytesMessage createBytesMessage()
            throws JMSException
    {
        return MQUtil.SESSION.createBytesMessage();
    }
    public static MapMessage createMapMessage()
            throws JMSException
    {
        return MQUtil.SESSION.createMapMessage();
    }
    public static Message createMessage()
            throws JMSException
    {
        return MQUtil.SESSION.createMessage();
    }
    public static ObjectMessage createObjectMessage()
            throws JMSException
    {
        return MQUtil.SESSION.createObjectMessage();
    }
    public static ObjectMessage createObjectMessage(Serializable serializable)
            throws JMSException
    {
        return MQUtil.SESSION.createObjectMessage(serializable);
    }
    public static StreamMessage createStreamMessage()
            throws JMSException
    {
        return MQUtil.SESSION.createStreamMessage();
    }
    public static TextMessage createTextMessage()
            throws JMSException
    {
        return MQUtil.SESSION.createTextMessage();
    }
    public static TextMessage createTextMessage(String text)
            throws JMSException
    {
        return MQUtil.SESSION.createTextMessage(text);
    }



    public static void createQueue(String name)
            throws JMSException
    {
        Destination destination = MQUtil.DESTINATIONS.put(name, MQUtil.SESSION.createQueue(name));
        MQUtil.MESSAGE_PRODUCERS.put(name, MQUtil.SESSION.createProducer(destination));
        MQUtil.MESSAGE_CONSUMERS.put(name, MQUtil.SESSION.createConsumer(destination));
    }
    public static void createTopic(String name)
            throws JMSException
    {
        Topic topic = MQUtil.TOPICS.put(name, MQUtil.SESSION.createTopic(name));
        MQUtil.TOPIC_PUBLISHERS.put(name, MQUtil.SESSION.createPublisher(topic));
        MQUtil.TOPIC_SUBSCRIBERS.put(name, MQUtil.SESSION.createSubscriber(topic));
    }


    public static MessageProducer getMessageProducer(String queue_name) {return MQUtil.MESSAGE_PRODUCERS.get(queue_name);}
    public static MessageConsumer getMessageConsumer(String queue_name) {return MQUtil.MESSAGE_CONSUMERS.get(queue_name);}
    public static TopicPublisher getTopicPublisher(String topic_name) {return MQUtil.TOPIC_PUBLISHERS.get(topic_name);}
    public static TopicSubscriber getTopicSubscriber(String topic_name) {return MQUtil.TOPIC_SUBSCRIBERS.get(topic_name);}


    public static void setMessageConsumerListener(String queue_name, MessageListener listener)
            throws JMSException
    {
        MQUtil.MESSAGE_CONSUMERS.get(queue_name).setMessageListener(listener);
    }
    public static void setTopicSubscriberListener(String topic_name, MessageListener listener)
            throws JMSException
    {
        MQUtil.TOPIC_SUBSCRIBERS.get(topic_name).setMessageListener(listener);
    }



    public static Message receiveQueueMessage(String queue_name)
            throws JMSException
    {
        MessageConsumer c = MQUtil.MESSAGE_CONSUMERS.get(queue_name);
        if (c == null)
            throw new NullPointerException("MessageConsumer is null");
        return c.receive();
    }
    public static Message receiveTopicMessage(String topic_name)
            throws JMSException
    {
        TopicSubscriber s = MQUtil.TOPIC_SUBSCRIBERS.get(topic_name);
        if (s == null)
            throw new NullPointerException("TopicSubscriber is null");
        return s.receive();
    }

    public static void sendQueueMessage(String queue_name, Message message)
            throws JMSException
    {
        MessageProducer p = MQUtil.MESSAGE_PRODUCERS.get(queue_name);
        if (p == null)
            throw new NullPointerException("MessageProducer is null");
        p.send(message);
    }

    public static void sendTopicMessage(String topic_name, Message message)
            throws JMSException
    {
        TopicPublisher p = MQUtil.TOPIC_PUBLISHERS.get(topic_name);
        if (p == null)
            throw new NullPointerException("TopicPublisher is null");
        p.send(message);
    }


}
